package cn.tedu.flinkbasic.datastream;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * 接收来自Kafka的数据并进行处理
 */
public class DataStreamDemo {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.获取数据源
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.65.161:9092");
        properties.setProperty("group.id", "test");
        DataStreamSource<String> source = env.addSource(new FlinkKafkaConsumer<>("flux", new SimpleStringSchema(), properties));
        //3.转化数据
        source.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                String[] split = value.split("\\|");
                return split.length == 16 && split[14].split("_").length == 3;
            }
        }).map(new MapFunction<String, Log>() {
            @Override
            public Log map(String value) throws Exception {
                String[] s = value.split("\\|");
                Log log = new Log();
                log.setUrl(s[0]);
                log.setUrlName(s[1]);
                log.setUvid(s[13]);
                log.setSsid(s[14].split("_")[0]);
                log.setSscount(s[14].split("_")[1]);
                log.setSstime(s[14].split("_")[2]);
                log.setCip(s[15]);
                log.setCount(1);

                return log;
            }
        }).keyBy("urlName").timeWindow(Time.days(1)).sum("count")
        //4.输出结果  zk  kafka  hadoop  flume  eclipse  idea
        .print();
        //5.触发执行:在DataStreamAPI中必须要加这一步.
        env.execute("DataStreamDemo");
    }
}
